package cn.gwm.flink.streaming.ods;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;

import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.flink.streaming.task.BaseTask;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.concurrent.TimeUnit;

/**
 * @ClassName EvEs13GpsToHive
 * @Author xzh
 * @Date 2022/11/18 14:38
 **/
public class EvEs13GpsToHive {
    public final static String CONSUMER_GROUP_ID_GPS_ES13 = "csonsumer-test-gps-es13";
    public final static String TOPIC_NAME_GPS_ES13 = "test_t5_gps_balei_signal_i_d";
    public final static String HIVE_PREFIX_ES13 = "test_es13_source";
    public final static String HIVE_PATH_ES13 = "test/test_es13_source";

    public static void addEs13ToHive(StreamExecutionEnvironment env) {
        final FlinkKafkaConsumer<String> kafkaConsumer = FlinkKafkaUtil.getConsumerEarLiest(
                CONSUMER_GROUP_ID_GPS_ES13,
                TOPIC_NAME_GPS_ES13
        );
        // 添加车型字段
       env.addSource(kafkaConsumer).map(JSONUtil::parseObj).returns(JSONObject.class)
                .addSink(StreamingFileSink.<JSONObject>forRowFormat(
                        new Path(ConfigLoader.get("hdfsUri") + "/external/data/" + HIVE_PATH_ES13),
                        new BaseTask.MySimpleStringEncoder<JSONObject>(BeanSource.SourceEnum.testEs13Gps)
                        //指定桶分配器
                ).withBucketAssigner(
                        new BucketAssigner<JSONObject, String>() {
                            @Override
                            public String getBucketId(JSONObject e, Context context) {
                                return "date" + "=" + e.getStr("date");
                            }

                            @Override
                            public SimpleVersionedSerializer getSerializer() {
                                return SimpleVersionedStringSerializer.INSTANCE;
                            }
                        }
                        //指定文件滚动策略
                ).withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(10))
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(1))
                                .withMaxPartSize(128 * 1024 * 1024)
                                .build()
                )
                        .withOutputFileConfig( OutputFileConfig.builder()
                                .withPartPrefix(HIVE_PREFIX_ES13)
                                .withPartSuffix(".txt")
                                .build())
                        .build());

    }

}
